package cn.jolyoulu.mqtt.service;


import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Import;
import org.springframework.stereotype.Component;
import cn.jolyoulu.mqtt.config.mqtt.EnableMqttClient;
import cn.jolyoulu.mqtt.config.mqtt.MqttClientPool;
import cn.jolyoulu.mqtt.config.mqtt.MqttProperties;

@Slf4j
@Component
@Import(MqttProperties.class)
@ConditionalOnBean(annotation = EnableMqttClient.class)
public class MqttSubTopicClient {

    private static MqttClient client = null;

    /**
     * 订阅
     */
    private static String subTopic = "testtopic/#";

    public MqttSubTopicClient(MqttProperties properties, MqttSubTopicService mqttSubTopicService) {
        //初始化一个客户端
        String clientId = properties.getNode() + "-MqttSubTopicClient";
        OnMessageCallback callback = new OnMessageCallback(clientId, properties,mqttSubTopicService);
        try {
            client = MqttClientPool.createClient(clientId, properties.getBroker(), MqttClientPool.jwtSecret, callback);
        } catch (MqttException e) {
            log.error("MQTT客户端[" + clientId + "]创建失败", e);
        }
    }



    public static class OnMessageCallback implements MqttCallbackExtended {

        /**
         * 客户端id
         */
        private final String clientId;

        /**
         * 环境
         */
        private final MqttProperties properties;

        /**
         * 处理消息订阅
         */
        private final MqttSubTopicService mqttSubTopicService;

        public OnMessageCallback(String clientId, MqttProperties properties, MqttSubTopicService mqttSubTopicService) {
            this.clientId = clientId;
            this.properties = properties;
            this.mqttSubTopicService = mqttSubTopicService;
        }

        public void connectionLost(Throwable cause) {
            // 连接丢失后，一般在这里面进行重连
            log.error("MQTT客户端[" + clientId + "] 与MQTT连接失败");
            OnMessageCallback callback = new OnMessageCallback(clientId, properties, mqttSubTopicService);
            //无限重试，直到成功
            while (true) {
                try {
                    MqttSubTopicClient.client.close(true);
                    MqttSubTopicClient.client = MqttClientPool.createClient(clientId, properties.getBroker(), MqttClientPool.jwtSecret, callback);
                    break;
                } catch (MqttException e) {
                    log.error("MQTT客户端[" + clientId + "]创建失败", e);
                }
            }
        }

        public void messageArrived(String topic, MqttMessage message) {
            // subscribe后得到的消息会执行到这里面
            mqttSubTopicService.callBack(topic, message);
        }

        public void deliveryComplete(IMqttDeliveryToken token) {
            log.info("MQTT客户端[{}]消息发送成功{}", client.getClientId(), token.isComplete());
        }

        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            //连接成功后回调
            try {
                client.subscribe(subTopic);
            } catch (MqttException e) {
                log.error("MQTT客户端[" + client.getClientId() + "]订阅消息失败", e);
            }
        }
    }
}
